package com.bigdata.warehouse.ads;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: com.bigdata.warehouse.ads
 * @Author: zhangleqing
 * @CreateTime: 2025-08-23  15:35
 * @Description: TODO
 * @Version: 1.0
 */
public class User {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 用户进出表
        tEnv.executeSql(
                "CREATE TABLE dws_owner_behavior (\n" +
                        "    ownerId INT COMMENT '业主ID',\n" +
                        "    owerSex STRING COMMENT '业主性别（1男/2女）',\n" +
                        "    owerAge INT COMMENT '业主年龄',\n" +
                        "    owerHomeplace STRING COMMENT '业主籍贯',\n" +
                        "    owerWorkplace STRING COMMENT '业主工作地',\n" +
                        "    community_id INT COMMENT '社区ID',\n" +
                        "    communityName STRING COMMENT '社区名称',\n" +
                        "    city_id INT COMMENT '城市ID',\n" +
                        "    district_id INT COMMENT '区域ID',\n" +
                        "    district_name STRING COMMENT '区域名称',\n" +
                        "    entrance_card_cnt INT COMMENT '刷卡开门次数',\n" +
                        "    entrance_pw_cnt INT COMMENT '密码开门次数',\n" +
                        "    entrance_face_cnt INT COMMENT '人脸开门次数',\n" +
                        "    entrance_remote_cnt INT COMMENT '远程开门次数',\n" +
                        "    stt TIMESTAMP(3) COMMENT '窗口开始时间',\n" +
                        "    edt TIMESTAMP(3) COMMENT '窗口结束时间',\n" +
                        "    ts TIMESTAMP(3) COMMENT '数据处理时间',\n" +
                        "    PRIMARY KEY (ownerId) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'upsert-kafka',\n" +
                        "    'topic' = 'dws_owner_behavior',\n" +
                        "    'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "    'key.format' = 'json',\n" +
                        "    'value.format' = 'json'\n" +
                        ")"
        );



/*
创建Doris表 sca_doris_export.dws_owner_behavior
CREATE TABLE dws_owner_behavior_doris (
    ownerId INT,
    owerSex STRING,
    owerAge INT,
    owerHomeplace STRING,
    owerWorkplace STRING,
    community_id INT,
    communityName STRING,
    city_id INT,
    district_id INT,
    district_name STRING,
    entrance_card_cnt INT,
    entrance_pw_cnt INT,
    entrance_face_cnt INT,
    entrance_remote_cnt INT,
    stt TIMESTAMP(3),
    edt TIMESTAMP(3),
    ts TIMESTAMP(3),
    PRIMARY KEY (ownerId) NOT ENFORCED
) WITH (
    'connector' = 'doris',
    'fenodes' = 'node102:8030',
    'table.identifier' = 'sca_doris_export.dws_owner_behavior',
    'username' = 'root',
    'password' = '123456',
    'sink.buffer-flush.max-rows' = '1',
    'sink.buffer-flush.interval' = '1000',
    'sink.max-retries' = '3',
    'sink.enable-delete' = 'true',
    'sink.properties.format' = 'json',
    'sink.enable.batch-mode' = 'true'
)
 */
        tEnv.executeSql(
                "CREATE TABLE dws_owner_behavior_doris (\n" +
                        "    ownerId INT,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    community_id INT,\n" +
                        "    communityName STRING,\n" +
                        "    city_id INT,\n" +
                        "    district_id INT,\n" +
                        "    district_name STRING,\n" +
                        "    entrance_card_cnt INT,\n" +
                        "    entrance_pw_cnt INT,\n" +
                        "    entrance_face_cnt INT,\n" +
                        "    entrance_remote_cnt INT,\n" +
                        "    stt TIMESTAMP(3),\n" +
                        "    edt TIMESTAMP(3),\n" +
                        "    ts TIMESTAMP(3),\n" +
                        "    PRIMARY KEY (ownerId) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'sca_doris_export.dws_owner_behavior',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.buffer-flush.max-rows' = '1',\n" +
                        "    'sink.buffer-flush.interval' = '1000',\n" +
                        "    'sink.max-retries' = '3',\n" +
                        "    'sink.enable-delete' = 'true',\n" +
                        "    'sink.properties.format' = 'json',\n" +
                        "    'sink.enable.batch-mode' = 'true'\n" +
                        ")"
        );

        tEnv.executeSql(
                "insert into dws_owner_behavior_doris " +
                        "select * from dws_owner_behavior"
        );

        tEnv.executeSql("select * from dws_owner_behavior_doris limit 10").print();

        //env.execute("User");
    }
}
